package cn._51doit.day06;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

/**
 * @create: 2021-10-21 23:08
 * @author: 今晚打脑斧先森
 * @program: EventTimeSlidingWindowDemo
 * @Description:
 *    先keyBy，再按照EventTime划分滑动动窗口
 *    设置延迟时间0
 *
 *    WaterMark = 产生WaterMark对应的DataStream每个分区最大的EventTime - 延迟时间
 *
 *    使用新的API提取EventTime生成WaterMark
 *
 *    小总结:
 *      滑动窗口是10秒,滑动步数是5秒
 *      在你输入数据的时候,就已经产生了窗口,直到你的数据超过滑动步数5秒的时候,就会触发已经创建的那个窗口
 *      这个窗口是-5秒到5秒 触发条件是5秒,因为用的新的APi,要是老的API那就是4999毫秒触发
 **/
public class EventTimeSlidingWindowDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);


        //5000,spark,3
        //7000,spark,3
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        SingleOutputStreamOperator<String> streamWithWaterMark = lines
                .assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            //第一个参数 5000,spark,3
                            @Override
                            public long extractTimestamp(String line, long l) {
                                String[] fields = line.split(",");
                                return Long.parseLong(fields[0]);
                            }
                        }));
        SingleOutputStreamOperator<Tuple2<String, Integer>> tpStream = streamWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {

                String[] split = value.split(",");
                return Tuple2.of(split[1], Integer.parseInt(split[2]));
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = tpStream.keyBy(tp -> tp.f0);
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = windowedStream.sum(1);

        res.print();
        env.execute();


    }
}
